package com.gxjtkyy.datadistributer.writer.rdbms.service;

import com.gxjtkyy.datadistributer.common.exception.ColumnParseException;
import com.gxjtkyy.datadistributer.writer.rdbms.config.WriterConfig;
import com.gxjtkyy.datadistributer.writer.rdbms.vo.DbOperation;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import javax.annotation.PostConstruct;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.CollectionUtils;
import org.springframework.util.backoff.FixedBackOff;

@Service
public class KafkaConsumerService {
   private static final Logger log = LoggerFactory.getLogger(KafkaConsumerService.class);
   private static final String ORA_00001_UNIQUE_CONSTRAINT_VIOLATED = "ORA-00001";
   private static final String ORA_12899_VALUE_TOO_LARGE = "ORA-12899";
   private static final String ORA_01438_VALUE_LARGER_THAN_SEPCIFIED_PRECISION = "ORA-01438";
   private static final String ORA_01400_CANNOT_INSERT_NULL = "ORA-01400";
   private final AbstractDbWriterService<?> dbWriterService;
   private final ConsumerFactory<Object, Object> consumerFactory;
   private final WriterConfig writerConfig;
   private final TransactionTemplate transactionTemplate;
   private final ExceptionMessageService exceptionMessageService;
   private final MetricService metricService;

   public KafkaConsumerService(AbstractDbWriterService<?> dbWriterService, ConsumerFactory<Object, Object> consumerFactory, WriterConfig writerConfig, TransactionTemplate transactionTemplate, ExceptionMessageService exceptionMessageService, MetricService metricService) {
      this.dbWriterService = dbWriterService;
      this.consumerFactory = consumerFactory;
      this.writerConfig = writerConfig;
      this.transactionTemplate = transactionTemplate;
      this.exceptionMessageService = exceptionMessageService;
      this.metricService = metricService;
   }

   @PostConstruct
   public void create() {
      this.writerConfig.getListenTopics().forEach((topic, value) -> {
         int concurrency = value == null ? 1 : value;

         for(int i = 0; i < concurrency; ++i) {
            this.createListener(topic);
         }

      });
   }

   private void createListener(String topic) {
      ContainerProperties containerProperties = new ContainerProperties(new String[]{topic});
      containerProperties.setMessageListener(new BatchMessageListener<String, String>() {
         public void onMessage(List<ConsumerRecord<String, String>> record) {
            consumerMessage(record, topic);
         }
      });
      //单条消费
//      containerProperties.setMessageListener(new MessageListener<String, String>() {
//         @Override
//         public void onMessage(ConsumerRecord<String, String> record) {
//            List<ConsumerRecord<String, String>> list = new ArrayList<>(Collections.singletonList(record));
//            consumerMessage(list, topic);
//         }
//      });
      ConcurrentMessageListenerContainer<String, String> concurrentMessageListenerContainer = new ConcurrentMessageListenerContainer(this.consumerFactory, containerProperties);
      FixedBackOff backOff = new FixedBackOff();
      backOff.setInterval(this.writerConfig.getBackOffInterval());
      concurrentMessageListenerContainer.setBatchErrorHandler(new RetryingBatchErrorHandler(backOff, (ConsumerRecordRecoverer)null));
      // 使用 RetryingErrorHandler 而不是 RetryingBatchErrorHandler
//      concurrentMessageListenerContainer.setErrorHandler(new ErrorHandler() {
//         @Override
//         public void handle(Exception e, ConsumerRecord<?, ?> consumerRecord) {
//            String errorMessage = getErrorMessage(e);
//         }
//      });
      concurrentMessageListenerContainer.start();
   }

   public void consumerMessage(List<ConsumerRecord<String, String>> data, String topic) {
      log.info("{} 写入数据库, size={}", topic, data.size());
      List<DbOperation> insertDbOperations = this.getInsertDbOperations(data);
      if (CollectionUtils.isEmpty(insertDbOperations)) {
         this.metricService.countSendToKafkaOk(topic, (long)data.size());
         return;
      }
      Status status = this.saveToDb(insertDbOperations, true);
//      Status status  = Status.SUCCESS;
      //批量失败时，按单条插入
      if (status != Status.SUCCESS) {
//         List<Status> results = (List)data.stream().map(this::consumerSingleMessage).collect(Collectors.toList());
         this.metricService.countSendToKafkaError(topic, (long)data.size());
//         if (results.stream().anyMatch((result) -> {
//            return Status.FAIL == result;
//         })) {
//            log.error("消息入库时发生异常, messages: {}", data.stream().map(ConsumerRecord::value).collect(Collectors.toList()));
//         }
         log.error("*************************消息主题:{}入库时发生异常****************************", topic);
         data.stream().map(ConsumerRecord::value).forEach(recordMessage -> log.error(recordMessage));
         log.error("=========================消息主题:{}入库时发生异常============================", topic);
//         log.error("消息入库时发生异常, messages: {}", data.stream().map(ConsumerRecord::value).collect(Collectors.toList()));
      } else {
         this.metricService.countSendToKafkaOk(topic, (long)data.size());
      }

   }

   private Status consumerSingleMessage(ConsumerRecord<String, String> consumerRecord) {
      String recordMessage = (String)consumerRecord.value();
      boolean insertOnly = false;
      boolean dropWhenValueTooLarge = false;
      boolean dropWhenCannotInsertNull = false;
      boolean dropWhenValueLargeThanSpecifiedPrecision = false;

      try {
         List<ConsumerRecord<String, String>> data = Collections.singletonList(consumerRecord);
         List<DbOperation> mergeDbOperations = this.getMergeDbOperations(data);
         if (mergeDbOperations.size() == 1) {
            DbOperation dbOperation = (DbOperation)mergeDbOperations.get(0);
            if (!dbOperation.isUpdatable()) {
               insertOnly = true;
            }

            if (dbOperation.isDropWhenValueTooLarge()) {
               dropWhenValueTooLarge = true;
            }

            if (dbOperation.isDropWhenCannotInsertNull()) {
               dropWhenCannotInsertNull = true;
            }

            if (dbOperation.isDropWhenValueLargeThanSpecifiedPrecision()) {
               dropWhenValueLargeThanSpecifiedPrecision = true;
            }
         }

         this.dbWriterService.writeDb(mergeDbOperations);
      } catch (ColumnParseException var10) {
         log.error("解析字段时发生错误, message={}", recordMessage);
         log.error("解析字段时发生错误", var10);
         this.saveErrorMessage(recordMessage, this.getErrorMessage(var10));
      } catch (Exception var11) {
         String errorMessage = this.getErrorMessage(var11);
         if (insertOnly && errorMessage.contains("ORA-00001")) {
            log.warn("违反唯一约束条件且该表设置为不可更新，跳过该记录，message={}", recordMessage);
            return Status.SUCCESS;
         }
         if (insertOnly && errorMessage.contains("错误: 重复键违反唯一约束")) {
            log.warn("违反唯一约束条件且该表设置为不可更新，跳过该记录，message={}", recordMessage);
            return Status.SUCCESS;
         }

         log.error("入库时发生错误, message={}", recordMessage);
         log.error("入库时发生错误", var11);
         if (dropWhenValueTooLarge && errorMessage.contains("ORA-12899")) {
            this.saveErrorMessage(recordMessage, errorMessage);
         } else if (dropWhenCannotInsertNull && errorMessage.contains("ORA-01400")) {
            this.saveErrorMessage(recordMessage, errorMessage);
         } else {
            if (!dropWhenValueLargeThanSpecifiedPrecision || !errorMessage.contains("ORA-01438")) {
               return Status.FAIL;
            }

            this.saveErrorMessage(recordMessage, errorMessage);
         }
      }

      return Status.SUCCESS;
   }

   private void saveErrorMessage(String recordMessage, String errorMessage) {
      log.error("写入错误消息表");

      try {
         this.exceptionMessageService.saveExceptionMessage(errorMessage, recordMessage);
      } catch (Exception var4) {
         log.error("写入错误消息表时发生错误", var4);
         throw var4;
      }
   }

   private List<DbOperation> getMergeDbOperations(List<ConsumerRecord<String, String>> data) {
      return (List)data.stream().map(ConsumerRecord::value).map((value) -> {
         try {
            return this.dbWriterService.generateMergeDbOperation(value);
         } catch (Exception var3) {
            log.error("生成数据库merge操作时发生异常", var3);
            log.error("data={}", value);
            throw new RuntimeException(var3);
         }
      }).filter(Objects::nonNull).collect(Collectors.toList());
   }

   private List<DbOperation> getInsertDbOperations(List<ConsumerRecord<String, String>> data) {
      return (List)data.stream().map(ConsumerRecord::value).map((value) -> {
         try {
//            return this.dbWriterService.generateInsertDbOperation(value);
            return this.dbWriterService.generateSaveOrUpdateDbOperation(value);
         } catch (Exception var3) {
            log.error("生成数据库insert操作时发生异常", var3);
            log.error("data={}", value);
//            throw new RuntimeException(var3);
            return null;
         }
      }).filter(Objects::nonNull).collect(Collectors.toList());
   }

   public Status saveToDb(List<DbOperation> dbOperations, boolean insert) {
      this.transactionTemplate.setTimeout(60);
      return (Status)this.transactionTemplate.execute((status) -> {
         try {
            this.dbWriterService.writeDb(dbOperations);
            return Status.SUCCESS;
         } catch (Exception var5) {
            if ("mysql".equalsIgnoreCase(this.writerConfig.getTargetDbType())) {
               if (!insert || this.getErrorMessage(var5).contains("Duplicate entry")) {
                  log.error("检测到mysql主键冲突，插入失败", var5);
               }
            } else if ("oracle".equalsIgnoreCase(this.writerConfig.getTargetDbType())) {
               if (!insert || !this.getErrorMessage(var5).contains("ORA-00001")) {
                  log.error("写入数据库时发生异常", var5);
               }
            } if ("postgresql".equalsIgnoreCase(this.writerConfig.getTargetDbType())) {
               if (!insert || this.getErrorMessage(var5).contains("错误: 重复键违反唯一约束")) {
                  log.error("检测到postgresql主键冲突，插入失败", var5);
                  DbOperation dbOperation = dbOperations.get(0);
                  boolean updatable = dbOperation.isUpdatable();
                  if (!updatable) {
                     log.info("主键冲突，跳过merge");
                     return Status.SUCCESS;
                  }
               }else {
                  log.error("写入数据库时发生异常", var5);
               }
            } else if (!insert || !this.getErrorMessage(var5).contains("ORA-00001")) {
               log.error("写入数据库时发生异常", var5);
            } else {
               log.error("写入数据库时发生异常", var5);
            }

            status.setRollbackOnly();
            return Status.FAIL;
         }
      });
   }

   private String getErrorMessage(Throwable throwable) {
      StringWriter stringWriter = new StringWriter();
      PrintWriter printWriter = new PrintWriter(stringWriter);

      String var4;
      try {
         throwable.printStackTrace(printWriter);
         var4 = stringWriter.toString();
      } catch (Throwable var7) {
         try {
            printWriter.close();
         } catch (Throwable var6) {
            var7.addSuppressed(var6);
         }

         throw var7;
      }

      printWriter.close();
      return var4;
   }

   public static enum Status {
      SUCCESS,
      FAIL;
   }
}
